package com.xyzq.simpson.hibbert.storage.influx;

import com.google.common.base.Splitter;
import com.xyzq.simpson.bart.client.BartClient;
import com.xyzq.simpson.bart.client.core.BartListener;
import com.xyzq.simpson.base.json.JSONArray;
import com.xyzq.simpson.base.json.JSONObject;
import com.xyzq.simpson.base.json.JSONVisitor;
import com.xyzq.simpson.base.json.core.IJSON;
import com.xyzq.simpson.base.model.core.IModule;
import com.xyzq.simpson.base.text.Text;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.UnsupportedEncodingException;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @Description: influxdb connection manager
 * @date 2020-3-5 8:47
 */
public class InfluxdbConnectionManager implements IModule {

    static Logger logger = LoggerFactory.getLogger(InfluxdbConnectionManager.class);

    static ConcurrentHashMap<Integer, InfluxDBConnection> influxConnsMap = new ConcurrentHashMap<>();

    static ConcurrentHashMap<String, Integer> shardingsRuleMap = new ConcurrentHashMap<>();

    public static ConcurrentHashMap<String, String> influxServiceMaps = new ConcurrentHashMap<>();

    private static volatile boolean initial = false;

    final static int influxCount = Integer.parseInt(BartClient.instance().
            fetch("simpson-hibbert-collector.influx.cluster.count"));

    private static InfluxdbConnectionManager instance;

    /**
     * 名字空间
     */
    protected String namespace;

    /**
     * 获取名字空间
     *
     * @return 名字空间
     */
    public String namespace() {
        return this.namespace;
    }

    /**
     * 设置名字空间
     *
     * @param namespace 名字空间
     */
    public void setNamespace(String namespace) {
        this.namespace = namespace;
    }

    public InfluxdbConnectionManager() {
        instance = this;
    }

    public static InfluxdbConnectionManager instance() {
        return instance;
    }

    @Override
    public synchronized boolean initialize() {
        if(initial) {
            return initial;
        }
        if(Text.isBlank(namespace)) {
            throw new RuntimeException("influxdb connection namespace is empty");
        }
        JSONObject iLinks = JSONObject.convertFromString(BartClient.instance().fetch(namespace));
        int i = 0;
        for(IJSON json : (JSONArray)iLinks.get("influxdbs")) {
            if(null != json) {
                JSONVisitor visitor = new JSONVisitor(json);
                String username = visitor.getString("username");
                String password = visitor.getString("password");
                String openurl = visitor.getString("openurl");
                String validUrl = !openurl.contains("http") ? "http://"+openurl : openurl;
                String database = visitor.getString("database");
                InfluxDBConnection influxDBConnection = new InfluxDBConnection(username,
                        password, validUrl, database, "");
                influxConnsMap.put(i, influxDBConnection);
                i++;
                if(i > influxCount) {
                    throw new RuntimeException("influxdb cluster set not fit ");
                }
            }
        }

        //commonservice
        commonService();

        //shardings rule
        shardingRule();

        return true;
    }

    private void commonService() {
        String commonservice = BartClient.instance().fetch("simpson-hibbert-service.report.influx.commonservice");
        doUploadCommonService(commonservice);
        BartClient.instance().listen("simpson-hibbert-service.report.influx.commonservice", new BartListener() {
            @Override
            public void onChange(String key, String value) {
                logger.info("commonservice config change, key = " + key + ", value = " + value);
                doUploadCommonService(value);
            }
        });
    }

    private void doUploadCommonService(String commonservice) {
        List<String> serviceMaps = Splitter.on(";").trimResults().omitEmptyStrings().splitToList(commonservice);
        for(String service : serviceMaps) {
            List<String> kv = Splitter.on(":").trimResults().omitEmptyStrings().splitToList(service);
            if(kv.size() == 2) {
                influxServiceMaps.put(kv.get(0), kv.get(1));
            }
        }
    }

    //指定分片
    private void shardingRule() {
        JSONObject iLinks = JSONObject.convertFromString(BartClient.instance().fetch("simpson-hibbert-collector.influx.shardings"));
        int i = 0;
        for(IJSON json : (JSONArray)iLinks.get("shardings")) {
            if(null != json) {
                JSONVisitor visitor = new JSONVisitor(json);
                String tables = visitor.getString("tables");
                List<String> tableList = Splitter.on(",").trimResults().omitEmptyStrings().splitToList(tables);
                for(String table : tableList) {
                    shardingsRuleMap.put(table, i);
                }
                i++;
                if(i > influxCount) {
                    throw new RuntimeException("influxdb shardings set not fit ");
                }
            }
        }
        initial = true;
    }

    /**
    * @Description:
    * @date 2020-3-5 10:05
    */
    @Override
    public void terminate() {
        if(influxConnsMap != null && influxConnsMap.size() > 0) {
            influxConnsMap.forEach((k,v) -> {
                if(Objects.nonNull(v)) {
                    v.close();
                }
            });
        }
    }

    public static int fetchShadingId(String type) {
        if(Text.isBlank(type)) {
            return 0;
        }
        byte[] bytes = new byte[0];
        try {
            bytes = type.getBytes("utf-8");
        }
        catch (UnsupportedEncodingException e) { }
        int i = 0;
        for(byte b : bytes) {
            i = i + (b & 0xFF);
        }
        return Math.abs(i % influxCount);
    }

    @Deprecated
    public static InfluxDBConnection fetchShadingConn(String type) {
        return influxConnsMap.get(fetchShadingId(type));
    }

    public static InfluxDBConnection fetchFixedShadingConn(String measurement) {
        if(!initial) {
            return null;
        }
        InfluxDBConnection res;
        if (shardingsRuleMap.containsKey(measurement)) {
            res = fetchShadingConn(shardingsRuleMap.get(measurement));
        } else {
            logger.warn("found unsetting measurement:" + measurement);
            return fetchShadingConn(fetchShadingId(measurement));
        }
        return res;
    }

    public static InfluxDBConnection fetchShadingConn(int index) {
        return influxConnsMap.get(index);
    }
}
